package practice.phase03.module04.zookeeper.distlock;

import org.I0Itec.zkclient.IZkDataListener;
import org.I0Itec.zkclient.ZkClient;

import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;

/**
 * DistClient.java
 * 抢锁
 * 创建临时序列节点，序号最小，则抢锁成功
 * 否则监听序号-1的节点，等待其结束通知
 * 获取到锁后，执行任务，任务完成后释放锁
 *
 * @author Alex Zhang
 */
public class DistClient {
    private ZkClient zk = new ZkClient("n1:2181,n2:2181");
    private String preNode;
    private CountDownLatch countDown;
    private String newLock;

    public void getDistLock() {
        String threadName = Thread.currentThread().getName();
        if (attemp()) {
            System.out.println(threadName + ": got lock");
        } else {
            // 需要等待，前任锁释放后，再次尝试获取锁
            System.out.println(threadName + ": wait for lock");
            waitForRelease();
            // 递归获取锁，结束条件为获取锁成功
            getDistLock();
        }
    }

    private boolean attemp() {
        String distLockNode = "/distlocks";
        if (null == newLock || newLock.isEmpty()) {
            newLock = zk.createEphemeralSequential(distLockNode + "/", "");
        }
        String newLockNum = newLock.replaceFirst(".+/", "");
        List<String> children = zk.getChildren(distLockNode);
        Collections.sort(children);
        String min = children.get(0);
        if (newLockNum.equals(min)) {
            return true;
        } else {
            int i = Collections.binarySearch(children, newLockNum);
            preNode = distLockNode + "/" + children.get(i - 1);
            waitForRelease();
            return false;
        }
    }

    private void waitForRelease() {
        IZkDataListener dataListener = new IZkDataListener() {
            @Override
            public void handleDataChange(String dataPath, Object data) throws Exception {
                //
            }

            @Override
            public void handleDataDeleted(String dataPath) throws Exception {
                countDown.countDown();
            }
        };
        zk.subscribeDataChanges(preNode, dataListener);
        if (zk.exists(preNode)) {
            countDown = new CountDownLatch(1);
            try {
                countDown.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        zk.unsubscribeDataChanges(preNode, dataListener);
    }

    public void releaseLock() {
        if (zk.exists(newLock)) {
            zk.delete(newLock);
        }
    }
}
